package raft

// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.

// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry

// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	"6.824/labgob"
	"bytes"
	"time"

	//	"bytes"
	"sync"
	"sync/atomic"
	//	"6.824/labgob"
	"6.824/labrpc"
)

// ApplyMsg
// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in part 2D you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh, but set CommandValid to false for these
// other uses.
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int
	// For 2D:
	SnapshotValid bool
	Snapshot      []byte
	SnapshotTerm  int
	SnapshotIndex int
}

type Log struct {
	Command  interface{}
	Term     int
	LogIndex int
}

// Raft
// A Go object implementing a single Raft peer.
type Raft struct {
	mu        sync.Mutex          // Lock to protect shared access to this peer's state
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted state
	me        int                 // this peer's index into peers[]
	dead      int32               // set by Kill()
	// Your data here (2A, 2B, 2C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.
	term              int
	state             int //0: follower; 1: candidate; 2: leader
	changeState       chan int
	votedFor          int
	logs              []Log
	applyCh           chan ApplyMsg
	commitIndex       int
	lastApplied       int
	lastIncludedTerm  int
	lastIncludedIndex int
	//leader
	nextIndex  []int
	matchIndex []int
}

// GetState
// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (term int, isLeader bool) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	term = rf.term
	isLeader = rf.state == 2
	return
}

// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
func (rf *Raft) persist() {
	// Your code here (2C).
	// Example:
	rf.mu.Lock()
	//fmt.Printf("%v persist\n", rf.me)
	defer rf.mu.Unlock()
	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	e.Encode(rf.term)
	e.Encode(rf.votedFor)
	e.Encode(rf.logs)
	e.Encode(rf.lastIncludedIndex)
	e.Encode(rf.lastIncludedTerm)
	data := w.Bytes()
	rf.persister.SaveRaftState(data)
}

// restore previously persisted state.
func (rf *Raft) readPersist(data []byte) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (2C).
	// Example:
	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)
	var term int
	var voteFor int
	var logs []Log
	var lastIncludedTerm int
	var lastIncludedIndex int
	if d.Decode(&term) != nil ||
		d.Decode(&voteFor) != nil ||
		d.Decode(&logs) != nil ||
		d.Decode(&lastIncludedTerm) != nil ||
		d.Decode(&lastIncludedIndex) != nil {
	} else {
		rf.term = term
		rf.votedFor = voteFor
		rf.logs = logs
		rf.lastIncludedTerm = lastIncludedTerm
		rf.lastIncludedIndex = lastIncludedIndex
		rf.commitIndex = rf.lastIncludedIndex
		rf.lastApplied = rf.lastIncludedIndex
	}
}

// CondInstallSnapshot A service wants to switch to snapshot.  Only do so if Raft hasn't
// have more recent info since it communicate the snapshot on applyCh.
func (rf *Raft) CondInstallSnapshot(lastIncludedTerm int, lastIncludedIndex int, snapshot []byte) bool {
	//不建议实现这个接口
	// Your code here (2D).
	return true
}

// Snapshot the service says it has created a snapshot that has
// all info up to and including index. this means the
// service no longer needs the log through (and including)
// that index. Raft should now trim its log as much as possible.
func (rf *Raft) Snapshot(index int, snapshot []byte) {
	// Your code here (2D).
	rf.mu.Lock()
	defer rf.mu.Unlock()
	if rf.lastIncludedIndex >= index {
		return
	}
	DPrintf("%v delete index %v", rf.me, index)
	for i, log := range rf.logs {
		if log.LogIndex == index {
			rf.lastIncludedIndex = index
			rf.lastIncludedTerm = log.Term
			rf.logs = rf.logs[i+1:]
			break
		}
	}
	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	e.Encode(rf.term)
	e.Encode(rf.votedFor)
	e.Encode(rf.logs)
	e.Encode(rf.lastIncludedIndex)
	e.Encode(rf.lastIncludedTerm)
	data := w.Bytes()
	rf.persister.SaveStateAndSnapshot(data, snapshot)
}

// Start the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	term, isLeader := rf.GetState()
	index := -1
	if !isLeader {
		return -1, -1, isLeader
	} else {
		//leader将新日志加入自己的log中
		rf.mu.Lock()
		lastIdx, _ := rf.getLastLogIndexAndTerm()
		index = lastIdx + 1
		newLog := Log{
			Command:  command,
			Term:     term,
			LogIndex: index,
		}
		rf.logs = append(rf.logs, newLog)
		DPrintf("%v receive %v %v\n", rf.me, index, command)
		go rf.heartBeats(term, rf.me, rf.commitIndex)
		rf.mu.Unlock()
		rf.persist()
		return index, term, isLeader
	}
}

func (rf *Raft) Kill() {

	rf.mu.Lock()
	defer rf.mu.Unlock()
	DPrintf("%v be killed\n", rf.me)
	DPrintf("node %v applyID %v commitID %v \n", rf.me, rf.lastApplied, rf.commitIndex)
	//for _, log := range rf.logs {
	//	DPrintf("%v ", log)
	//}
	//DPrintf("\n")
	atomic.StoreInt32(&rf.dead, 1)
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

func (rf *Raft) applyLog() {
	for !rf.killed() {
		time.Sleep(20 * time.Millisecond)
		//If commitIndex > lastApplied: increment lastApplied, apply
		//log[lastApplied] to state machine (§5.3)
		rf.mu.Lock()
		LogsToApply := make([]Log, 0)
		for _, log := range rf.logs {
			if log.LogIndex > rf.lastApplied && log.LogIndex <= rf.commitIndex {
				LogsToApply = append(LogsToApply, log)
				DPrintf("%v apply %v\n", rf.me, log.LogIndex)
			}
		}
		rf.lastApplied = rf.commitIndex
		rf.mu.Unlock()
		for _, log := range LogsToApply {
			rf.applyCh <- ApplyMsg{
				CommandValid: true,
				Command:      log.Command,
				CommandIndex: log.LogIndex,
			}
		}
	}
}

// Make
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me
	rf.changeState = make(chan int, 30)
	rf.votedFor = -1
	rf.applyCh = applyCh
	rf.logs = make([]Log, 0)
	rf.nextIndex = make([]int, len(peers))
	rf.matchIndex = make([]int, len(peers))
	// Your initialization code here (2A, 2B, 2C).
	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())
	// start ticker goroutine to start elections
	go rf.ticker()
	go rf.applyLog()
	return rf
}
